package com.socket.websocket.netty;



import com.socket.batch.deal.AbstractDeal;
import com.socket.entity.GpsRecord;
import com.socket.entity.PushData;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.CharsetUtil;
import org.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.InputStreamReader;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.*;

import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;

/**
 * A HTTP server which serves Web Socket requests at:
 *
 * http://localhost:8080/websocket
 *
 * Open your browser at http://localhost:8080/, then the demo page will be
 * loaded and a Web Socket connection will be made automatically.
 *
 * This server illustrates support for the different web socket specification
 * versions and will work with:
 *
 * <ul>
 * <li>Safari 5+ (draft-ietf-hybi-thewebsocketprotocol-00)
 * <li>Chrome 6-13 (draft-ietf-hybi-thewebsocketprotocol-00)
 * <li>Chrome 14+ (draft-ietf-hybi-thewebsocketprotocol-10)
 * <li>Chrome 16+ (RFC 6455 aka draft-ietf-hybi-thewebsocketprotocol-17)
 * <li>Firefox 7+ (draft-ietf-hybi-thewebsocketprotocol-10)
 * </ul>
 *
 * netty方式
 */
public class WebSocketServer implements IWebSocketServer {
    private static final Logger logger = LoggerFactory.getLogger(WebSocketServer.class);

    //定时器
    private ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
    //计数
    private static int i = 1;
    /**
     * 保存所有WebSocket连接
     */
    private Map<ChannelId, Channel> channelMap = new ConcurrentHashMap<ChannelId, Channel>();

    private WebSocketServerHandshaker handshaker;

    public Map<ChannelId, Channel> getChannelMap() {
        return channelMap;
    }

    public void setChannelMap(Map<ChannelId, Channel> channelMap) {
        this.channelMap = channelMap;
    }

    private int port;

    public WebSocketServer(int port) {
        this.port = port;
    }

    public static WebSocketServer build(int port) throws Exception {
        System.out.println("WebsocketServer 开始创建" + "----port:" + port);
        WebSocketServer server = new WebSocketServer(port);
        return server;
    }

    public void run() throws Exception {

        EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap(); // (2)
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class) // (3)
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            ChannelPipeline pl = ch.pipeline();
                            // 保存该Channel的引用
                            channelMap.put(ch.id(), ch);
                            logger.info("new channel {}", ch);
                            ch.closeFuture().addListener(new ChannelFutureListener() {

                                public void operationComplete(ChannelFuture future) throws Exception {
                                    logger.info("channel close {}", future.channel());
                                    // Channel 关闭后不再引用该Channel
                                    channelMap.remove(future.channel().id());
                                }
                            });

                            pl.addLast(new HttpServerCodec());
                            pl.addLast(new HttpObjectAggregator(64 * 1024));
                            pl.addLast(new ChunkedWriteHandler());
                            pl.addLast(new WebSocketServerHandler(WebSocketServer.this));
                            //pl.addLast(new WebSocketServerProtocolHandler("/ws"));



                        }
                    })  //(4)
                    .option(ChannelOption.SO_BACKLOG, 128)          // (5)
                    .childOption(ChannelOption.SO_KEEPALIVE, true); // (6)

            System.out.println("WebsocketServer 启动了" + "----port:" + port);

            // 绑定端口，开始接收进来的连接
            ChannelFuture f = b.bind(port).sync(); // (7)
            if (i == 1) {//保证只需起来一次
                service.scheduleAtFixedRate(new PublishThread(), 0, 3, TimeUnit.SECONDS);
            }
            i++;
            // 等待服务器  socket 关闭 。
            // 在这个例子中，这不会发生，但你可以优雅地关闭你的服务器。
            f.channel().closeFuture().sync();

        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();

            System.out.println("WebsocketServer 关闭了");
        }
    }

    public void handleHttpRequest(ChannelHandlerContext ctx,
                                  FullHttpRequest req) throws Exception {

        // 如果HTTP解码失败，返回HHTP异常
        if (!req.decoderResult().isSuccess()
                || (!"websocket".equals(req.headers().get("Upgrade")))&&(!"Websocket".equals(req.headers().get("Upgrade")))) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1,
                    BAD_REQUEST));
            return;
        }

        // 构造握手响应返回，本机测试
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                "ws://localhost:8091/websocket", null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory
                    .sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }

    }

    public void handleWebSocketFrame(ChannelHandlerContext ctx,
                                     WebSocketFrame frame) {

        // 判断是否是关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(),
                    (CloseWebSocketFrame) frame.retain());
            return;
        }
        // 判断是否是Ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(
                    new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        // 判断是否是Pong消息
        if (frame instanceof PongWebSocketFrame) {
            ctx.channel().write(
                    new PingWebSocketFrame(frame.content().retain()));
            return;
        }
        // 本例程仅支持文本消息，不支持二进制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(String.format(
                    "%s frame types not supported", frame.getClass().getName()));
        }

        logger.info("recieve TextWebSocketFrame from channel {}", ctx.channel());

    }

    private static void sendHttpResponse(ChannelHandlerContext ctx,
                                         FullHttpRequest req, FullHttpResponse res) {
        // 返回应答给客户端
        if (res.getStatus().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(),
                    CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
            HttpUtil.setContentLength(res, res.content().readableBytes());
        }

        // 如果是非Keep-Alive，关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }


    //线程任务
    class PublishThread implements Runnable {

        @Override
        public void run() {
            //while(true) {
                try {
                    logger.info("websocket PublishThread 开始推送!");
                    //System.out.println("websocket PublishThread 开始推送!");
                    BlockingQueue<GpsRecord> blockingQueue = AbstractDeal.mapQueue.get("socket");
                    //测试用
//                    if (blockingQueue == null){
//                        String s = "123";
//                        // 发给其他所有channel
//                        for (Channel ch : channelMap.values()) {
//                            ch.write(new TextWebSocketFrame(s));
//                            ch.flush();
//                            logger.info("write text[{}] to channel {}", s, ch);
//                        }
//                    }
                    if (blockingQueue.size() > 0) {
                        GpsRecord gpsRecord = null;
                        JSONObject pushData = new JSONObject();
                        JSONArray dataArray = new JSONArray();
                        pushData.put("vehicles", dataArray);
                        JSONObject vehicle = new JSONObject();
                        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        while (blockingQueue.size() > 0) {
                            gpsRecord = blockingQueue.poll();
                            vehicle.put("nm", gpsRecord.getVehicleNo());
                            vehicle.put("lng", gpsRecord.getLon());
                            vehicle.put("lat", gpsRecord.getLat());
                            System.out.println("gpsRecord.getOnline()"+gpsRecord.getOnline());
                            vehicle.put("online", gpsRecord.getOnline());
                            vehicle.put("sp", gpsRecord.getSpeed());
                            vehicle.put("createDate", dateFormat.format(gpsRecord.getCreateDate()));
                            dataArray.put(vehicle);
                        }
                        String s = JSONObject.valueToString(pushData);
                        logger.info("websocket 推送队列有数据 {}, 开始读取", s);
                        //System.out.println("websocket 推送队列有数据 " + s + "开始读取");
                        // 发给其他所有channel
                        for (Channel ch : channelMap.values()) {
                            ch.write(new TextWebSocketFrame(s));
                            ch.flush();
                            logger.info("write text[{}] to channel {}", s, ch);
                        }
                    } /*else {
                        String s = "";
                        // 发给其他所有channel
                        for (Channel ch : channelMap.values()) {
                            ch.write(new TextWebSocketFrame(s));
                            ch.flush();
                            logger.info("write text[{}] to channel {}", s, ch);
                        }
                    }*/
                } catch (Exception e) {
                    e.getStackTrace();
                }
            }

        //}
    }
}
